package net.maergaiyun.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import net.maergaiyun.model.ShortLinkVisitStatsDO;
import net.maergaiyun.util.KafkaUtil;
import net.maergaiyun.util.MyClickHouseSink;
import net.maergaiyun.util.TimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple9;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

@Slf4j
public class DwsShortLinkVisitStatsApp {

    /**
     * 宽表
     */
    public static final String SHORT_LINK_SOURCE_TOPIC = "dwm_link_visit_topic";

    public static final String SHORT_LINK_SOURCE_GROUP = "dws_link_visit_group";

    /**
     * uv数据流
     */
    public static final String UNIQUE_VISIT_SOURCE_TOPIC = "dwm_unique_visit_topic";

    public static final String UNIQUE_VISIT_SOURCE_GROUP = "dws_unique_visit_group";


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1、获取多个数据流
        FlinkKafkaConsumer<String> shortLinkSource = KafkaUtil.getKafkaConsumer(SHORT_LINK_SOURCE_TOPIC, SHORT_LINK_SOURCE_GROUP);
        DataStreamSource<String> shortLinkDS = env.addSource(shortLinkSource);

        FlinkKafkaConsumer<String> uniqueVisitSource = KafkaUtil.getKafkaConsumer(UNIQUE_VISIT_SOURCE_TOPIC, UNIQUE_VISIT_SOURCE_GROUP);
        DataStreamSource<String> uniqueVisitDS = env.addSource(uniqueVisitSource);

        // 2、结构转换, uniqueVisitDS, shortLinkDS
        SingleOutputStreamOperator<ShortLinkVisitStatsDO> shortLinkMapDS = shortLinkDS.map(new MapFunction<String, ShortLinkVisitStatsDO>() {
            @Override
            public ShortLinkVisitStatsDO map(String value) throws Exception {
                ShortLinkVisitStatsDO visitStatsDO = parseVisitStats(value);
                // 点击量为1
                visitStatsDO.setPv(1L);
                visitStatsDO.setUv(0L);
                return visitStatsDO;
            }
        });
        SingleOutputStreamOperator<ShortLinkVisitStatsDO> uniqueVisitMapDS = uniqueVisitDS.map(new MapFunction<String, ShortLinkVisitStatsDO>() {
            @Override
            public ShortLinkVisitStatsDO map(String value) throws Exception {
                ShortLinkVisitStatsDO visitStatsDO = parseVisitStats(value);
                visitStatsDO.setPv(0L);
                // 独立IP访客量为1
                visitStatsDO.setUv(1L);
                return visitStatsDO;
            }
        });

        // 3、多流合并(合并相同结构的流)
        DataStream<ShortLinkVisitStatsDO> unionDS = shortLinkMapDS.union(uniqueVisitMapDS);

        // 4、设置waterMark
        SingleOutputStreamOperator<ShortLinkVisitStatsDO> watermarkDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
                // 指定允许乱序延迟最大3秒
                .<ShortLinkVisitStatsDO>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                // 指定事件时间列, 毫秒
                .withTimestampAssigner((event, timestamp) -> {
                    log.info(event.toString());
                    event.getVisitTime();
                    return event.getVisitTime();
                }));

        // 5、多维度, 多字段分组
        // code, referer, isNew, province, city, ip, browserName, os, deviceType
        KeyedStream<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>> keyedStream = watermarkDS.keyBy(new KeySelector<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>>() {
            @Override
            public Tuple9<String, String, Integer, String, String, String, String, String, String> getKey(ShortLinkVisitStatsDO value) throws Exception {
                return Tuple9.of(value.getCode(), value.getReferer(), value.getIsNew(),
                        value.getProvince(), value.getCity(), value.getIp(),
                        value.getBrowserName(), value.getOs(), value.getDeviceType()
                );
            }
        });

        // 6、开窗, 10秒一次数据插入到ck
        WindowedStream<ShortLinkVisitStatsDO, Tuple9<String, String, Integer, String, String, String, String, String, String>, TimeWindow> windowedStream =
                keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        // 7、聚合统计(补充统计起止时间)
        SingleOutputStreamOperator<Object> reduceDS = windowedStream.reduce(new ReduceFunction<ShortLinkVisitStatsDO>() {
            @Override
            public ShortLinkVisitStatsDO reduce(ShortLinkVisitStatsDO v1, ShortLinkVisitStatsDO v2) throws Exception {
                v1.setPv(v1.getPv() + v2.getPv());
                v1.setUv(v1.getUv() + v2.getUv());
                return v1;
            }
        }, new ProcessWindowFunction<ShortLinkVisitStatsDO, Object, Tuple9<String, String, Integer, String, String, String, String, String, String>, TimeWindow>() {
            @Override
            public void process(Tuple9<String, String, Integer, String, String, String, String, String, String> tuple,
                                Context context, Iterable<ShortLinkVisitStatsDO> iterable, Collector<Object> collector) throws Exception {
                for (ShortLinkVisitStatsDO shortLinkVisitStatsDO : iterable) {
                    String startTime = TimeUtil.formatWithTime(context.window().getStart());
                    String endTime = TimeUtil.formatWithTime(context.window().getEnd());
                    shortLinkVisitStatsDO.setStartTime(startTime);
                    shortLinkVisitStatsDO.setEndTime(endTime);
                    collector.collect(shortLinkVisitStatsDO);
                }
            }
        });

        reduceDS.print(">>>>>>>>");

        // 8、写入ClickHouse
        String sql = "insert into visit_stats values(?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?)";
        reduceDS.addSink(MyClickHouseSink.getJdbcSink(sql));

        env.execute();
    }

    private static ShortLinkVisitStatsDO parseVisitStats(String value) {
        JSONObject jsonObj = JSON.parseObject(value);
        ShortLinkVisitStatsDO visitStatsDO = ShortLinkVisitStatsDO.builder().code(jsonObj.getString("code")).accountNo(jsonObj.getLong("accountNo")).visitTime(jsonObj.getLong("visitTime")).referer(jsonObj.getString("referer")).isNew(jsonObj.getInteger("isNew")).udid(jsonObj.getString("udid"))
                // 地理位置信息
                .province(jsonObj.getString("province")).city(jsonObj.getString("city")).isp(jsonObj.getString("isp")).ip(jsonObj.getString("ip"))
                // 设备信息
                .browserName(jsonObj.getString("browserName")).os(jsonObj.getString("os")).osVersion(jsonObj.getString("osVersion")).deviceType(jsonObj.getString("deviceType")).deviceManufacture(jsonObj.getString("deviceManufacture")).build();
        return visitStatsDO;
    }
}
